package rocket;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.alibaba.rocketmq.common.message.MessageQueue;

import java.io.UnsupportedEncodingException;

/**
 * 推送消息的消费
 * @author wsz
 * @date 2022/2/22 18:38
 **/
public class PushConsumer {

    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_grp_wsz");
        consumer.setNamesrvAddr("192.168.174.138:9876");
        consumer.subscribe("tp_demo1", "*");
        // 消息监听器：有消息推送过来进行消费
        consumer.setMessageListener((MessageListenerConcurrently) (messageExts, context) -> {
            final MessageQueue messageQueue = context.getMessageQueue();
            System.out.println(messageQueue);
            for (MessageExt msg : messageExts) {
                try {
                    System.out.println(new String(msg.getBody(), "utf8"));
                } catch (UnsupportedEncodingException e) {
                    e.printStackTrace();
                }
            }
            // 消息消费成功
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            // 消费失败
//            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        });

        consumer.start();
        Thread.sleep(30000);
        consumer.shutdown();
    }
}
